use std::{
    collections::VecDeque,
    fs::{remove_file, File, OpenOptions},
    io::Write,
    os::unix::fs::symlink,
    path::Path,
    sync::{
        atomic::{AtomicUsize, Ordering},
        Arc,
    },
    time::Duration,
};

use bytes::BufMut;
use chrono::{Datelike, Local, Timelike};
use parking_lot::Mutex;
use tokio::{
    sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender},
    time::timeout,
};

use crate::error::*;
use crate::{runtime::RuntimeManager, utils::time::get_now_msec};

use super::LogOutput;

const DEF_BUF_SIZE: usize = 1024 * 1024 * 4;
const DEF_PENDING_BUFS: usize = 8;
const DEF_FLUSH_INTERVAL_MS: u64 = 1000;

#[allow(dead_code)]
#[derive(PartialEq, PartialOrd, Clone, Copy, Debug)]
#[repr(u8)]
pub enum Rolling {
    None = 1,
    Minute = 2,
    Hour = 3,
    Day = 4,
}

pub struct FileOutputOptions {
    pub buf_size: usize,
    pub max_pending_bufs: usize,
    pub flush_interval_ms: u64,
    pub rolling: Rolling,
}

impl Default for FileOutputOptions {
    fn default() -> Self {
        FileOutputOptions {
            buf_size: DEF_BUF_SIZE,
            max_pending_bufs: DEF_PENDING_BUFS,
            flush_interval_ms: DEF_FLUSH_INTERVAL_MS,
            rolling: Rolling::None,
        }
    }
}

impl FileOutputOptions {
    pub fn valid(mut self) -> Self {
        if self.buf_size < DEF_BUF_SIZE {
            self.buf_size = DEF_BUF_SIZE;
        }

        if self.max_pending_bufs < DEF_PENDING_BUFS {
            self.max_pending_bufs = DEF_PENDING_BUFS;
        }

        if self.flush_interval_ms < DEF_FLUSH_INTERVAL_MS {
            self.flush_interval_ms = DEF_FLUSH_INTERVAL_MS;
        }

        self
    }
}

#[allow(dead_code)]
struct FileOutputInner {
    opts: Arc<FileOutputOptions>,
    activate_buf: Option<Vec<u8>>,
    pending_bufs: VecDeque<Vec<u8>>,
    free_bufs: Vec<Vec<u8>>,
    last_switch_ms: u64,
    stopped: bool,
}

impl FileOutputInner {
    fn new(opts: Arc<FileOutputOptions>) -> Self {
        let max_pending_bufs = opts.max_pending_bufs;
        let buf_size = opts.buf_size;

        let mut inner = FileOutputInner {
            opts,
            activate_buf: None,
            pending_bufs: VecDeque::with_capacity(max_pending_bufs),
            free_bufs: Vec::with_capacity(max_pending_bufs),
            last_switch_ms: get_now_msec(),
            stopped: false,
        };

        for _ in 0..max_pending_bufs {
            inner.free_bufs.push(Vec::with_capacity(buf_size));
        }

        inner
    }
}

#[derive(PartialEq, PartialOrd, Clone, Copy, Debug)]
#[repr(u8)]
enum NotifyMsg {
    Stop = 1,
    Flush = 2,
}

struct FileHandle {
    file_path: String,
    real_path: String,
    log_file: File,
}

#[allow(dead_code)]
pub struct FileOutput {
    opts: Arc<FileOutputOptions>,
    loss: AtomicUsize,
    sender: UnboundedSender<NotifyMsg>,
    inner: Arc<Mutex<FileOutputInner>>,
    file_handle: Arc<Mutex<FileHandle>>,
}

fn open_real_log_file(file_path: &str, real_path: &str) -> Result<File> {
    let file = OpenOptions::new()
        .append(true)
        .create(true)
        .open(real_path)
        .if_err(ErrorType::OpenError, real_path.to_string())?;

    if file_path == real_path {
        return Ok(file);
    }

    let link = Path::new(&file_path);
    if link.is_file() {
        _ = remove_file(link);
    }

    symlink(real_path, file_path)
        .if_err(ErrorType::OpenError, "symlink failed")?;

    Ok(file)
}

async fn file_output_flush_loop(
    opts: Arc<FileOutputOptions>,
    inner: Arc<Mutex<FileOutputInner>>,
    file_handle: Arc<Mutex<FileHandle>>,
    mut receiver: UnboundedReceiver<NotifyMsg>,
) {
    let mut stop = false;
    let flush_duartion = Duration::from_millis(opts.flush_interval_ms);
    let mut pending_bufs = VecDeque::with_capacity(opts.max_pending_bufs);

    loop {
        let receive_envet = receiver.recv();
        let receive_envet = timeout(flush_duartion, receive_envet);

        if let Some(msg) = receive_envet.await.unwrap_or_default() {
            if msg == NotifyMsg::Stop {
                stop = true;
            }
        }

        // 判断log是否需要rolling
        {
            let mut guard = file_handle.lock();
            let real_path = real_path_gen(&guard.file_path, opts.rolling);
            if real_path != guard.real_path {
                if let Ok(new_file) =
                    open_real_log_file(&real_path, &guard.file_path)
                {
                    guard.real_path = real_path;
                    guard.log_file = new_file;
                }
            }
        }

        {
            let now = get_now_msec();
            let mut guard = inner.lock();
            if guard.last_switch_ms + opts.flush_interval_ms < now || stop {
                if let Some(buf) = guard.activate_buf.take() {
                    guard.pending_bufs.push_back(buf);
                }
            }

            while let Some(v) = guard.pending_bufs.pop_front() {
                pending_bufs.push_back(v);
            }
        }

        {
            let mut guard = file_handle.lock();
            for v in pending_bufs.iter_mut() {
                _ = guard.log_file.write_all(v);
                v.clear();
            }
        }

        {
            if !pending_bufs.is_empty() {
                let mut guard = inner.lock();
                while let Some(v) = pending_bufs.pop_front() {
                    guard.free_bufs.push(v);
                }
            }
        }

        if stop {
            break;
        }
    }

    inner.lock().stopped = true;
}

fn real_path_gen(file_path: &str, rolling: Rolling) -> String {
    let now_time = Local::now();

    match rolling {
        Rolling::None => file_path.to_string(),
        Rolling::Minute => {
            format!(
                "{}.{:04}{:02}{:02}{:02}{:02}",
                file_path,
                now_time.year(),
                now_time.month(),
                now_time.day(),
                now_time.hour(),
                now_time.minute()
            )
        }
        Rolling::Hour => {
            format!(
                "{}.{:04}{:02}{:02}{:02}",
                file_path,
                now_time.year(),
                now_time.month(),
                now_time.day(),
                now_time.hour()
            )
        }
        Rolling::Day => {
            format!(
                "{}.{:04}{:02}{:02}",
                file_path,
                now_time.year(),
                now_time.month(),
                now_time.day()
            )
        }
    }
}

#[allow(dead_code)]
impl FileOutput {
    pub fn new(
        opts: Option<FileOutputOptions>,
        file_path: String,
    ) -> Result<Self> {
        let opts = Arc::new(opts.unwrap_or_default());
        let (sender, receiver) = unbounded_channel();
        let inner = Arc::new(Mutex::new(FileOutputInner::new(opts.clone())));

        let real_path = real_path_gen(&file_path, opts.rolling);
        let log_file = open_real_log_file(&real_path, &file_path)?;
        let file_handle = Arc::new(Mutex::new(FileHandle {
            file_path,
            real_path,
            log_file,
        }));

        // start flush loop task.
        let inner_clone = inner.clone();
        let opts_clone = opts.clone();
        let file_handle_clone = file_handle.clone();
        RuntimeManager::get_worker_runtime()
            .get_handle()
            .spawn(async {
                file_output_flush_loop(
                    opts_clone,
                    inner_clone,
                    file_handle_clone,
                    receiver,
                )
                .await;
            });

        Ok(FileOutput {
            opts,
            loss: AtomicUsize::new(0),
            sender,
            inner,
            file_handle,
        })
    }
}

impl LogOutput for FileOutput {
    fn print(&self, msg: &str) {
        let buf_size = self.opts.buf_size;

        // log is too large.
        if msg.len() > buf_size / 2 {
            self.loss.fetch_add(1, Ordering::Relaxed);
            return;
        }

        let mut guard = self.inner.lock();
        if let Some(buf) = guard.activate_buf.as_mut() {
            if buf.len() + msg.len() < buf_size {
                buf.put_slice(msg.as_bytes());
                return;
            }

            let buf = guard.activate_buf.take().unwrap();
            guard.pending_bufs.push_back(buf);
            // notify flush loop.
            _ = self.sender.send(NotifyMsg::Flush);
        }

        match guard.free_bufs.pop() {
            Some(buf) => {
                guard.activate_buf = Some(buf);
            }
            None => {
                self.loss.fetch_add(1, Ordering::Relaxed);
                return;
            }
        }

        let buf = guard.activate_buf.as_mut().unwrap();
        buf.put_slice(msg.as_bytes());
    }

    fn flush(&self) {
        _ = self.sender.send(NotifyMsg::Flush);
    }

    // 该函数会导致线程阻塞. 直到异步任务处理完毕.
    // 一般只用在worker exit流程中.
    fn stop(&self) {
        _ = self.sender.send(NotifyMsg::Stop);
        let sleep_duration = Duration::from_millis(10);

        loop {
            if self.inner.lock().stopped {
                return;
            }
            std::thread::sleep(sleep_duration);
        }
    }
}
